package com.ellison.zookeeperdemo.queue;

import org.I0Itec.zkclient.ZkClient;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 *
 * 流量不大，数据不多的情况下使用zk队列
 *       大致范围在百万以上就不可使用了。
 *
 * @Classname FifoTest
 * @Description TODO
 * @Author xx
 * Date 2021/6/19 21:28
 * Version 1.0
 */
public class FifoTest {

    private volatile boolean stop = true;

    public static void main(String[] args) {
        new FifoTest().start();
    }

    private void start() {
        ZkClient zkClient = new ZkClient("119.3.228.104:2181");
        FIFOQueue queue = new FIFOQueue(zkClient, "/queue");

        new Thread(new Provider(queue)).start();
        new Thread(new Consumer(queue)).start();

        for (int i = 0; i < 100; i++) {
            if (i == 90) {
                new Thread(() -> {
                    try {
                        TimeUnit.SECONDS.sleep(3);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    stop = false;
                }).start();
            }
        }

    }

    class Consumer implements Runnable {

        private FIFOQueue queue;

        public Consumer(FIFOQueue queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            while (stop) {
                try {
                    TimeUnit.MILLISECONDS.sleep(100);
                    System.out.println(Thread.currentThread().getName() + "--> 成功poll数据！！" + queue.poll());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            System.out.println("=========消费者停止了!!");
        }
    }

    class Provider implements Runnable {

        private FIFOQueue queue;

        public Provider(FIFOQueue queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            while (stop) {
                try {
                    queue.push(UUID.randomUUID());
                    TimeUnit.MILLISECONDS.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            System.out.println("=========生产者停止了!!");
        }
    }
}
